package com.zyf.interceptor;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author Malegod_xiaofei
 * @create 2024-08-18-21:04
 */
public class TypeInterceptor implements Interceptor {

    // 声明一个集合用于存放拦截器处理后的事件
    private List<Event> addHeaderEvents;

    @Override
    public void initialize() {
        // 初始化
        addHeaderEvents = new ArrayList<>();
    }


    // 单个事件处理方法
    @Override
    public Event intercept(Event event) {
        // 1.获取 header & body
        Map<String, String> headers = event.getHeaders();
        String body = new String(event.getBody());

        // 2.根据 body 中是否包含“zyf”添加不同的头信息
        if (body.contains("zyf")) {
            headers.put("type", "zyf");
        } else {
            headers.put("type", "other");
        }

        // 3.返回数据
        return event;
    }

    // 批量事件处理方法
    @Override
    public List<Event> intercept(List<Event> events) {
        // 1.清空集合
        addHeaderEvents.clear();

        // 2.遍历events
        for (Event event : events) {
            addHeaderEvents.add(intercept(event));
        }

        // 3.返回数据
        return addHeaderEvents;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {
        public Interceptor build() {
            return new TypeInterceptor();
        }

        public void configure(Context context) {
        }
    }
}
